package org.example.rxJava;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Emitter;
import rx.Observable;
import rx.functions.Action1;
import rx.schedulers.Schedulers;

import java.util.concurrent.Future;

@Slf4j
public class SchedulerDemo {
    @Test
    public void testScheduler() throws Exception{
        Observable<String> observable = Observable.create(new Action1<Emitter<String>>() {
            @Override
            public void call(Emitter<String> emitter){
                try {
                    for (int i = 0; i < 5; i++) {
                        log.info("produce->{}", i);
                        // 生产一个消费一个
                        emitter.onNext(String.valueOf(i));
                    }
                    Thread.sleep(10000);
                    emitter.onCompleted();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, Emitter.BackpressureMode.NONE);

        // 异步
        observable
                .observeOn(Schedulers.newThread())
                .subscribe(s -> {
            log.info("consumr->{}", s);
        });
        log.info("consumer completed");
    }

    @Test
    public void countDemo() {
        String[] items = {"one", "two", "three", "four"};
        Observable.from(items).count()
                .observeOn(Schedulers.newThread())
                .subscribe((s)->{
            log.info("count->{}", s);
        });
        log.info("countDemo completed");
    }

    @Test
    public void testObserveOn() throws Exception{
        Future<Integer> future = Observable.from(new String[]{"1", "2", "3", "4", "5"})
                .count()
                .toBlocking().toFuture();
        log.info("result->{}", future.get());
    }
}
